package cn.tedu.rabbitmq.test;

import com.rabbitmq.client.*;
import org.junit.Before;
import org.junit.Test;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

public class Consumer {
    private Channel channel;
    @Before
    public void initChannel() throws IOException, TimeoutException {
        // 通过java客户端建立和rabbitmq的连接
        ConnectionFactory factory = new ConnectionFactory();
        // 给工厂提供连接属性 ip:port
        factory.setHost("192.168.8.100");
        factory.setPort(5672);
        // 权限账号 admin/123456
        factory.setUsername("admin");
        factory.setPassword("123456");
        // 虚拟主机：每个连接工厂由一个程序管理，程序所属的环境 连接的虚拟主机virtual_host不同
        // 实现不同环境的数据隔离 默认就是/
        factory.setVirtualHost("/");
        Connection connection = factory.newConnection();
        // 可以从长连接获取短链接
        channel = connection.createChannel();
    }
    @Test
    public void consumer01() throws IOException {
        // 为消息消费处理准备一个回调对象DeliverCallback;
        DeliverCallback deliver = new DeliverCallback() {
            /**
             *
             * @param consumerTag 处理消息的消费者标识
             * @param message 消息队列传递过来的消息对象 包含发送的消息所有数据
             *
             */
            public void handle(String consumerTag, Delivery message) {
                System.out.println("消费者01接收到消息");
                // 从message中获取 消息内容
                byte[] body = message.getBody();
                String msg=new String(body);
                // 消息来源交换机 消息路由键
                message.getEnvelope().getExchange();
                message.getEnvelope().getRoutingKey();
                // 消息携带的属性 现在是null
                message.getProperties();
                System.out.println("收到消息数据:"+msg);
            }
        };
        CancelCallback cancel = new CancelCallback() {
            /**
             *
             * 消费者 停止监听 拒绝消息 断开连接 调用的方法
             * 一般是在消费者 停止消费的时候回调 释放一些消费者持有的资源
             */
            public void handle(String consumerTag) {
                System.out.println("消费者01停止监听");
            }
        };
        // 调用客户端消费方法
        /*
        *  String queue: 队列名称,当前连接 对象监听消费的队列是谁
        *  boolean autoAck: 是否自动确认 true自动确认 false手动确认
        *  消费方法 处理是异步线程,消费结果 消费状态 异步回调
        * */
        channel.basicConsume("queue1",true,deliver,cancel);
        // 保证回调 可以看到处理结果 线程阻塞等待 死循环
        while(true);
    }
    @Test
    public void consumer02() throws IOException {
        DeliverCallback deliver = (consumerTag, message) -> {
            System.out.println("消费者02接收到消息");
            byte[] body = message.getBody();
            String msg = new String(body);
            System.out.println("收到消息数据:"+msg);
        };
        CancelCallback cancel = consumerTag -> System.out.println("消费者02停止监听");
        channel.basicConsume("queue1",true,deliver,cancel);
        //保证回调 可以看到处理结果 线程阻塞等待 死循环
        while(true);
    }
    @Test
    public void consumer03() throws IOException {
        DeliverCallback deliver= (consumerTag, message) -> {
            System.out.println("消费者03接收到消息");
            byte[] body = message.getBody();
            String msg=new String(body);
            System.out.println("收到消息数据:"+msg);
            //手动确认 根据消费结果 确认 可以重新投递
            /*channel.basicAck(message.getEnvelope().getDeliveryTag(),false);*/
            //手动确认 消费失败 重新投递
            channel.basicNack(message.getEnvelope().getDeliveryTag(),false,true);
            //直接拒绝 不需重新投递 消息去哪了? 去了死信队列
            /*channel.basicReject(message.getEnvelope().getDeliveryTag(),false);*/
        };
        CancelCallback cancel = consumerTag -> System.out.println("消费者03停止监听");
        channel.basicConsume("queue1",false,deliver,cancel);
    }
}
